package websocket

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"sync"
	"time"

	"github.com/gorilla/websocket"
	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/threading"
)

type AckType int

const (
	NoAck AckType = iota
	OnlyAck
	RigorAck
)

func (t AckType) Tostring() string {
	switch t {
	case OnlyAck:
		return "OnlyAck"
	case RigorAck:
		return "RigorAck"
	}
	return "NoAck"
}

type Server struct {
	*threading.TaskRunner
	sync.RWMutex
	opt            *serverOption
	authentication Authentication
	connToUser     map[*Conn]string
	Usertoconn     map[string]*Conn
	routers        map[string]HandleFunc
	addr           string
	pattern        string
	upgrader       websocket.Upgrader
	Discover       Discover
	listenOn       string
	logx.Logger
}

func NewServer(addr string, opts ...ServerOptions) *Server {
	opt := newServerOption(opts...)
	s := &Server{
		opt:            &opt,
		routers:        make(map[string]HandleFunc),
		Usertoconn:     map[string]*Conn{},
		connToUser:     map[*Conn]string{},
		addr:           addr,
		pattern:        opt.pattern,
		upgrader:       websocket.Upgrader{},
		Logger:         logx.WithContext(context.Background()),
		authentication: opt.Authentication,
		TaskRunner:     threading.NewTaskRunner(opt.concurrency),
		Discover:       opt.discover,
		listenOn:       addr,
	}
	return s
}

func (s *Server) GetAddr() string {
	return s.addr
}

func (s *Server) addConn(conn *Conn, req *http.Request) {
	uid := s.authentication.UserId(req)
	s.RWMutex.Lock()
	defer s.RWMutex.Unlock()

	//如果重复登陆，关闭原来的连接，使用新连接
	if c, ok := s.Usertoconn[uid]; ok {
		c.Close()
	}

	s.Usertoconn[uid] = conn
	s.connToUser[conn] = uid

}

func (s *Server) GetConn(uid string) *Conn {
	s.RWMutex.RLock()
	defer s.RWMutex.RUnlock()
	conn := s.Usertoconn[uid]
	return conn
}

func (s *Server) GetUserId(conn *Conn) string {
	s.RWMutex.RLock()
	defer s.RWMutex.RUnlock()
	UserId := s.connToUser[conn]
	return UserId
}

func (s *Server) GetConns(uid ...string) []*Conn {
	if len(uid) == 0 {
		return nil
	}
	s.RWMutex.RLock()
	defer s.RWMutex.RUnlock()
	conns := make([]*Conn, 0)
	for _, id := range uid {
		conns = append(conns, s.GetConn(id))
	}
	return conns
}

func (s *Server) GetUsers(conns ...*Conn) []string {

	s.RWMutex.RLock()
	defer s.RWMutex.RUnlock()

	var res []string
	if len(conns) == 0 {
		// 获取全部
		res = make([]string, 0, len(s.connToUser))
		for _, uid := range s.connToUser {
			res = append(res, uid)
		}
	} else {
		// 获取部分
		res = make([]string, 0, len(conns))
		for _, conn := range conns {
			res = append(res, s.connToUser[conn])
		}
	}

	return res
}

func (s *Server) Close(conn *Conn) error {
	err := conn.Close()

	s.RWMutex.Lock()
	defer s.RWMutex.Unlock()

	uid := s.connToUser[conn]
	if uid == "" {
		return nil
	}
	delete(s.connToUser, conn)
	delete(s.Usertoconn, uid)
	return err
}
func (s *Server) ServerWs(w http.ResponseWriter, r *http.Request) {
	defer func() {
		if r := recover(); r != nil {
			s.Errorf("server handler ws recover err %v", r)
		}
	}()

	conn := NewConn(s, w, r)
	if conn == nil {
		return
	}

	if !s.authentication.Auth(w, r) {
		s.Send(&Message{
			FrameType: FrameData,
			Data:      []byte(fmt.Sprintf("无权访问")),
		}, conn)
		//conn.WriteMessage(websocket.TextMessage, []byte("不具备访问权限"))
		conn.Close()
		return
	}
	//记录连接
	s.addConn(conn, r)
	logx.Info("BandUser TO Redis :", s.GetUsers(conn)[0])
	s.Discover.BandUser(s.GetUsers(conn)[0])
	//s.Info("记录连接记录连接记录连接记录连接记录连接记录连接")
	//处理连接
	go s.handlerConn(conn)
}

func (s *Server) Start() {
	http.HandleFunc(s.pattern, s.ServerWs)
	s.Info(http.ListenAndServe(s.addr, nil))

}

func (s *Server) Stop() {
	fmt.Println("stop server")
}

func (s *Server) AddRouters(rs []Route) {
	for _, v := range rs {
		s.routers[v.Method] = v.Handler
	}
}

func (s *Server) AddRoute(rs Route) {
	s.routers[rs.Method] = rs.Handler
}

// 根据连接对象执行任务处理
func (s *Server) handlerConn(conn *Conn) {

	uids := s.GetUsers(conn)
	conn.Uid = uids[0]
	go s.handlerWrite(conn)

	if s.opt.ack != NoAck {
		go s.readAck(conn)
	}
	for {
		_, msg, err := conn.ReadMessage()
		if err != nil {
			s.Errorf("websocket conn read message err %v", err)

			//关闭连接对象
			s.Close(conn)
			return
		}
		var message Message
		err = json.Unmarshal(msg, &message)
		if err != nil {
			s.Send(NewErrMessage(err), conn)
			continue
		}

		//todo :给客户端回复ACK

		if s.opt.ack != NoAck && message.FrameType != FrameNoAck { //需要ack
			//将消息加入队列中
			s.Infof("conn msg write in msgMq %v", message)
			conn.appendMsgMq(&message)
		} else {
			conn.messageCh <- &message
		}

	}
}

func (s *Server) readAck(conn *Conn) {
	send := func(msg *Message, conn *Conn) error {
		tempDelay := time.Duration(conn.readMessages[0].ErrCount) * time.Duration(conn.readMessages[0].ErrCount) * 500 * time.Millisecond
		if max := 3 * time.Second; tempDelay > max {
			tempDelay = max
		}
		time.Sleep(tempDelay)
		err := s.Send(msg, conn)
		if err == nil {
			return nil
		}
		s.Errorf("message ack OnlyAck send err:%v message:%v ", err, msg)
		conn.readMessages[0].ErrCount++
		conn.msgMu.Unlock()

		return err

	}

	for {
		select {
		case <-conn.done:
			s.Infof("连接已关闭")
			return
		default:
		}

		conn.msgMu.Lock()
		if len(conn.readMessages) == 0 { //暂无消息，睡眠等待
			conn.msgMu.Unlock()
			time.Sleep(100 * time.Millisecond)
			continue
		}

		message := conn.readMessages[0]
		if message.ErrCount > s.opt.maxReSendCounts {
			s.Infof("超过最大重发次数,message:%v,ackType:%v,maxReSendCount:%v", message, s.opt.ack.Tostring(), s.opt.maxReSendCounts)

			delete(conn.readMessagesSeq, message.Id)
			conn.readMessages = conn.readMessages[1:]
			conn.msgMu.Unlock()
			continue
		}

		//根据ack确认策略选择处理方式
		switch s.opt.ack {
		case OnlyAck:
			if err := send(&Message{
				FrameType: FrameAck,
				AckSeq:    message.AckSeq + 1,
				AckTime:   time.Now(),
				Id:        message.Id,
			}, conn); err != nil {
				continue
			}
			conn.readMessages = conn.readMessages[1:]
			conn.msgMu.Unlock()
			conn.messageCh <- message
			s.Infof("message ack OnlyAck send success msgId:%v", message.Id)
		case RigorAck:
			if message.AckSeq == 0 {
				conn.readMessages[0].AckSeq++
				conn.readMessages[0].AckTime = time.Now()
				conn.readMessages[0].ErrCount++
				ackMsg := Message{
					FrameType: FrameAck,
					AckSeq:    message.AckSeq,
					Id:        message.Id,
					AckTime:   message.AckTime,
					ErrCount:  message.ErrCount,
				}
				if s.opt.maxReSendCounts > message.ErrCount {
					if err := send(&ackMsg, conn); err != nil { //这里的锁可能有问题
						continue
					}
				}

				conn.msgMu.Unlock()
				s.Infof("message Ack RigorAck send msgId:%v,seq:%v,time:%v", ackMsg.Id, ackMsg.AckSeq, ackMsg.AckTime.Unix())
				continue
			}

			//服务端向客户端已发送第一次确认，需要客户端再次确认
			msgSeq := conn.readMessagesSeq[message.Id]
			if msgSeq.AckSeq > message.AckSeq { //客户端确认成功,可以处理业务了
				conn.readMessages = conn.readMessages[1:]
				conn.msgMu.Unlock()
				conn.messageCh <- message
				s.Infof("message Ack RigorAck success msgId:%v", message.Id)
				continue
			}

			//处理出问题了，先看看是否超时
			val := s.opt.ackTimeOut - time.Since(message.AckTime)
			if !message.AckTime.IsZero() && val <= 0 {
				s.Errorf("message ack RigorAck TimeOut,msgId:%v", message.Id)
				//超时了放弃该消息
				delete(conn.readMessagesSeq, message.Id)
				conn.readMessages = conn.readMessages[1:]
				conn.msgMu.Unlock()
				continue
			}

			conn.msgMu.Unlock()
			if val > 0 && val > 300*time.Millisecond {
				if s.opt.maxReSendCounts > message.ErrCount {
					message.ErrCount++
					if err := send(&Message{
						FrameType: FrameAck,
						AckSeq:    message.AckSeq,
						AckTime:   time.Now(),
						Id:        msgSeq.Id,
						ErrCount:  message.ErrCount,
					}, conn); err != nil {
						continue
					}
				}

			}
			time.Sleep(300 * time.Millisecond)
		}
	}

}

// 任务处理
func (s *Server) handlerWrite(conn *Conn) {
	for {
		select {
		case <-conn.done:
			return
		case message := <-conn.messageCh:
			switch message.FrameType {
			case FramePing:
				s.Send(&Message{
					FrameType: FramePing,
				}, conn)
			case FrameData, FrameNoAck:

				if hd, ok := s.routers[message.Method]; ok {
					hd(s, conn, message)
				} else {
					s.Send(&Message{
						FrameType: FrameData,
						Data:      []byte(fmt.Sprintf("不存在该执行方法Method:%v", message.Method)),
					}, conn)
					//conn.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf("不存在该执行方法Method:%v", message.Method)))
				}
			}
			//处理完一条任务后,从map中删除
			if s.opt.ack != NoAck {
				conn.msgMu.Lock()
				delete(conn.readMessagesSeq, message.Id)
				conn.msgMu.Unlock()
			}
		}

	}
}

func (s *Server) SendByUserId(msg interface{}, recvIds ...string) error {
	if len(recvIds) == 0 {
		return nil
	}

	return s.Send(msg, s.GetConns(recvIds...)...)
}

func (s *Server) Send(msg interface{}, conns ...*Conn) error {
	if len(conns) == 0 {
		return nil
	}

	data, err := json.Marshal(msg)
	if err != nil {
		return err
	}

	for _, conn := range conns {
		err := conn.WriteMessage(websocket.TextMessage, data)
		if err != nil {
			return err
		}
	}
	return nil
}
